package comsumer

import (
	"errors"
	"strconv"
	"sync"
	"time"

	"github.com/upmio/horus-agent/model"
	"github.com/upmio/horus-polymerizer/log"
	"github.com/upmio/horus-polymerizer/queue"

	"database/sql"
	"fmt"
	_ "github.com/go-sql-driver/mysql"
)

//EventComsumerManager is exported
type EventComsumerManager struct {
	curnum int
	maxnum int
	lock   *sync.Mutex
	//mysql source
	source string
}

//NewEventComsumerManager  is exported
func NewEventComsumerManager(source string, maxnum int) *EventComsumerManager {
	return &EventComsumerManager{
		curnum: 0,
		maxnum: maxnum,
		lock:   new(sync.Mutex),
		source: source,
	}
}

func (c *EventComsumerManager) getMysqlConnectedDB() (*sql.DB, error) {
	db, err := sql.Open("mysql", c.source)
	if err != nil {
		return nil, err
	}

	if err := db.Ping(); err != nil {
		return nil, fmt.Errorf("mysql ping fail:%s", err.Error())
	}

	return db, nil
}

//InitAndStartManager  is exported
func (c *EventComsumerManager) InitAndStartManager() {
	log.Info(" Start event comsumers and eventmanager")
	for i := 0; i < c.maxnum; i++ {
		go c.comsumerHandle()
	}
	go c.manager()
}

func getEventKey(event *model.EventMetric) string {
	documentID := event.EndPoint + ":" + event.Metric + ":" + strconv.Itoa(int(event.CreateStamp/60))
	return documentID
}

func (c *EventComsumerManager) addComsumer() error {
	c.lock.Lock()
	defer c.lock.Unlock()
	c.curnum++

	if c.curnum > c.maxnum {
		return errors.New("biger than maxnum")
	}

	return nil
}

func (c *EventComsumerManager) delComsumer() {
	c.lock.Lock()
	defer c.lock.Unlock()
	c.curnum--
}

func (c *EventComsumerManager) manager() {
	counter := 0
	time.Sleep(2 * 60 * time.Second)
	for {
		for i := c.curnum; i < c.maxnum; i++ {
			log.Info("restart one event comsumer")
			go c.comsumerHandle()
		}
		if counter%30 == 0 {
			log.Infof("the eventqueue infos: curlen:%d ,maxlen:%d", queue.GetEventCurLen(), queue.GetEventMaxLen())
		}

		counter++
		time.Sleep(2 * 60 * time.Second)
	}

}

func (c *EventComsumerManager) insertEvent(db *sql.DB, event *model.EventMetric) error {

	//TODO : if event exit return nil

	stmt, err := db.Prepare("INSERT tb_event SET id=?,alarm=?,send=?,obj_name=?," +
		"metric=?,value=?,level=?,createstamp=?,obj_container_name=?,obj_host_name=?," +
		"obj_type=?")

	if err != nil {
		return err
	}

	if _, err := stmt.Exec(event.Id, event.Alarm, false,
		event.EndPoint, event.Metric, event.Value,
		event.Level, event.CreateStamp,
		event.ObjContainerName, event.ObjHostName, event.ObjType); err != nil {

		return err
	}

	return nil
}

func (c *EventComsumerManager) comsumerHandle() {

	//operate bulket  may occur panic
	defer func() {
		if err := recover(); err != nil {
			log.Error("event comsumer occur panic : ", err)
		}
		c.delComsumer()
	}()

	if err := c.addComsumer(); err != nil {
		log.Warn("start event comsumer conflict:", err, " curlen:", c.curnum, "maxlen:", c.maxnum)
		return
	}

	db, err := c.getMysqlConnectedDB()
	if err != nil {
		log.Errorf("connect mysql  fail:%s", err.Error())
		return
	}

	defer db.Close()

	log.Info("the event comsumer start")

	breakCount := 0
	for {

		eventslice, _ := queue.GetEventSlice()
		log.Debug("the eventslice len is", len(eventslice))

		for _, event := range eventslice {
			if event == nil {
				log.Debug("event equal nil")
				continue
			}

			if err := c.insertEvent(db, event); err != nil {

				log.Errorf("insert event to mysql event table fail:%s", err.Error())

				if queue.GetEventCurLen()*2 < queue.GetEventMaxLen() {
					queue.PutEvent(event)
					log.Info("reput the event:%v to event queue", event)
				}

				breakCount++
				if breakCount > 4 {
					break
				}

			} else {
				log.Debugf("%v:write event to mysql ok", event)

			}

		}

		if breakCount > 4 {
			break
		}

		if err := db.Ping(); err != nil {
			log.Error("mysql ping fail")
			break
		}
	}
	log.Warn("the event comsumer stop")
}
